package com.asiainfo.dacp.datastash.broker.util;

import java.util.List;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache.StartMode;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.asiainfo.dacp.datastash.broker.core.BrokerContext;
import com.google.common.collect.Lists;

public class ZKUtil {
	private static Logger LOG = LoggerFactory.getLogger(ZKUtil.class);

	public static final int DEFAULT_SESSION_TIMEOUT = 60000;// zooKeeper
															// session的超时时间
	public static final int DEFAULT_CONN_TIMEOUT = 30000;// 连接zooKeeper的超时时间

	private static CuratorFramework client;

	static {
		client = CuratorFrameworkFactory.newClient(BrokerContext.ZK_ADDRESS, DEFAULT_SESSION_TIMEOUT,
				DEFAULT_CONN_TIMEOUT, new RetryNTimes(2000, 20000));
		client.start();
	}
	
	public static InterProcessMutex getLock(String lockPath){
		return new InterProcessMutex(client, lockPath);
	}

	public static List<String> children(String znode) {
		try {
			return client.getChildren().forPath(znode);
		} catch (Exception e) {
			LOG.error("获取[" + znode + "]节点children失败！", e);
		}
		return Lists.newArrayList();
	}

	public static void createP(String znode) {
		createP(znode, null);
	}

	public static void createE(String znode) {
		createE(znode, null);
	}

	public static void createP(String znode, String zdata) {
		create(znode, zdata, CreateMode.PERSISTENT);
	}

	public static void createE(String znode, String zdata) {
		create(znode, zdata, CreateMode.EPHEMERAL);
	}

	public static void create(String znode, String zdata, CreateMode mode) {
		try {
			if (null != exists(znode)) {
				return;
			}
			if (null != zdata) {
				client.create().creatingParentsIfNeeded().withMode(mode).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
						.forPath(znode, zdata.getBytes());
			} else {
				client.create().creatingParentsIfNeeded().withMode(mode).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
						.forPath(znode);
			}
		} catch (Exception e) {
			LOG.error("创建[" + znode + "]节点失败！", e);
		}
	}

	public static Stat exists(String znode) {
		Stat stat = null;
		try {
			stat = client.checkExists().forPath(znode);
		} catch (Exception e) {
			LOG.error("获取节点[" + znode + "]信息失败！", e);
		}
		return stat;
	}

	public static byte[] data(String path) throws Exception {
		return client.getData().forPath(path);
	}

	public static PathChildrenCache watchPathChild(final String nodePath, PathChildrenCacheListener listener)
			throws Exception {
		final PathChildrenCache childrenCache = new PathChildrenCache(client, nodePath, true);
		childrenCache.start(StartMode.POST_INITIALIZED_EVENT);
		childrenCache.getListenable().addListener(listener);
		return childrenCache;
	}

}
